package com.yl.util;

import com.yl.constant.Const;
import com.yl.entity.MultiDataEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class MysqlUtil {

    /**
     * 获取分流标签
     */
    public static List<OutputTag<MultiDataEntity>> getCalTags(ParameterTool params) throws Exception {
        List<OutputTag<MultiDataEntity>> calTags = new ArrayList<>();
        Connection connection = getConnection(params);
        PreparedStatement ps = connection.prepareStatement(Const.SQL_QUOTA_CAL);
        ResultSet rs = ps.executeQuery();
        while (rs.next()) {
            String calKey = SUtil.fmt(rs.getString(1), rs.getString(2), rs.getString(3));
            calTags.add(FlinkUtil.getTag(calKey));
        }
        return calTags;
    }

    /**
     * 连接数据库
     */
    public static Connection getConnection(ParameterTool params) throws Exception {
        Class.forName(params.get(Const.MYSQL_DRIVER));
        return DriverManager.getConnection(
                params.get(Const.MYSQL_URL), params.get(Const.MYSQL_USERNAME), params.get(Const.MYSQL_PASSWORD));
    }

}
